package org.stayfool.client.handler;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.UnsupportedMessageTypeException;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.stayfool.client.event.EventKey;
import org.stayfool.client.event.EventType;
import org.stayfool.client.event.MqttEvent;
import org.stayfool.client.session.SessionManager;
import org.stayfool.client.util.ChannelUtil;
import org.stayfool.client.util.message.FixHeaderUtil;

/**
 * @author stayfool
 */
public class MqttClientHandler extends ChannelInboundHandlerAdapter {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        MqttMessage message = (MqttMessage) msg;
        switch (message.fixedHeader().messageType()) {
            case CONNACK:
                processConnAck(ctx, (MqttConnAckMessage) message);
                break;
            case PUBLISH:
                processPublish(ctx, (MqttPublishMessage) message);
                break;
            case PUBACK:
                processPubAck(ctx, (MqttPubAckMessage) message);
                break;
            case PUBREC:
                processPubRec(ctx, (MqttPubAckMessage) message);
                break;
            case PUBREL:
                processPubRel(ctx, (MqttPubAckMessage) message);
                break;
            case PUBCOMP:
                processPubComp(ctx, (MqttPubAckMessage) message);
                break;
            case SUBACK:
                processSubAck(ctx, (MqttSubAckMessage) message);
                break;
            case UNSUBACK:
                processUnsbAck(ctx, (MqttUnsubAckMessage) message);
                break;
            case PINGRESP:
                processPingResp(ctx, message);
                break;
            default:
                throw new UnsupportedMessageTypeException("Unacceptable Message Type");
        }
    }

    private void processConnAck(ChannelHandlerContext ctx, MqttConnAckMessage message) {

        triggerUserEvent(ctx, EventType.CONNECT_COMPLETE, message);

        if (message.variableHeader().connectReturnCode().equals(MqttConnectReturnCode.CONNECTION_ACCEPTED)) {
            triggerUserEvent(ctx, EventType.CONNECT_SUCCESS, message);
            logger.info("{} connect success", ChannelUtil.clientId(ctx.channel()));
        } else {
            triggerUserEvent(ctx, EventType.CONNECT_FAILURE, message);
            logger.info("{} connect failure", ChannelUtil.clientId(ctx.channel()));
        }
    }

    private void processPublish(ChannelHandlerContext ctx, MqttPublishMessage message) {

        triggerUserEvent(ctx, EventType.MESSAGE_ARRIVE, message);
        MqttQoS qos = message.fixedHeader().qosLevel();

        if (qos.equals(MqttQoS.AT_LEAST_ONCE)) {
            MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBACK);
            MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().messageId());
            MqttPubAckMessage pubAck = new MqttPubAckMessage(fixedHeader, variableHeader);
            ctx.channel().writeAndFlush(pubAck);
        } else if (qos.equals(MqttQoS.EXACTLY_ONCE)) {
            MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBREC);
            MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().messageId());
            MqttPubAckMessage pubRec = new MqttPubAckMessage(fixedHeader, variableHeader);
            ctx.channel().writeAndFlush(pubRec);
        }

        if (message.fixedHeader().isRetain()) {
            SessionManager.getSession(ChannelUtil.clientId(ctx.channel())).retainMessage(message);
        }

        ByteBuf bb = message.payload().duplicate();
        byte[] msg = new byte[bb.readableBytes()];
        bb.readBytes(msg);
        logger.info("accept message : topic-{}; content-{}", message.variableHeader().topicName(), new String(msg, CharsetUtil.UTF_8));
    }

    private void processPubAck(ChannelHandlerContext ctx, MqttPubAckMessage message) {
        triggerUserEvent(ctx, EventType.PUBLISH_SUCCESS, message);
        logger.info("publish success : {}", ChannelUtil.clientId(ctx.channel()));
    }

    private void processPubRec(ChannelHandlerContext ctx, MqttPubAckMessage message) {
        MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBREL);
        MqttPubAckMessage pubRel = new MqttPubAckMessage(fixedHeader, message.variableHeader());
        ctx.channel().writeAndFlush(pubRel);
    }

    private void processPubRel(ChannelHandlerContext ctx, MqttPubAckMessage message) {
        MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBCOMP);
        MqttPubAckMessage pubComp = new MqttPubAckMessage(fixedHeader, message.variableHeader());
        ctx.channel().writeAndFlush(pubComp);
    }

    private void processPubComp(ChannelHandlerContext ctx, MqttPubAckMessage message) {
        triggerUserEvent(ctx, EventType.PUBLISH_SUCCESS, message);
        logger.info("publish success : {}", ChannelUtil.clientId(ctx.channel()));
    }

    private void processSubAck(ChannelHandlerContext ctx, MqttSubAckMessage message) {
        boolean success = true;
        if (message.payload().grantedQoSLevels().isEmpty())
            success = false;
        if (success) {
            for (int qos : message.payload().grantedQoSLevels()) {
                if (MqttQoS.FAILURE.equals(MqttQoS.valueOf(qos))) {
                    success = false;
                    break;
                }
            }
        }
        if (success) {
            triggerUserEvent(ctx, EventType.SUBSCRIBE_SUCCESS, message);
            logger.info("subscribe success : {} ", ChannelUtil.clientId(ctx.channel()));
        } else {
            triggerUserEvent(ctx, EventType.SUBSCRIBE_FAILURE, message);
            logger.info("subscribe failure : {} ", ChannelUtil.clientId(ctx.channel()));
        }
    }

    private void processUnsbAck(ChannelHandlerContext ctx, MqttUnsubAckMessage message) {
        triggerUserEvent(ctx, EventType.UNSUBSCRIBE_SUCCESS, message);
        logger.info("unsubscribe success : {} ", ChannelUtil.clientId(ctx.channel()));
    }

    private void processPingResp(ChannelHandlerContext ctx, MqttMessage message) {
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        logger.error("bad thing happened : ", cause);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        triggerUserEvent(ctx, EventType.DIS_CONNECT, null);
        logger.info("lose connection : {}", ChannelUtil.clientId(ctx.channel()));
    }

    private void triggerUserEvent(ChannelHandlerContext ctx, EventType type, MqttMessage message) {
        EventKey key = new EventKey(type, ChannelUtil.clientId(ctx.channel()));
        MqttEvent event = new MqttEvent(key, message);
        ctx.fireUserEventTriggered(event);

        logger.debug("trigger user event : {}", key);
    }
}
